package com.bff.gaia.streaming.api.xjoin.core.side;
/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


import com.bff.gaia.api.java.typeutils.RowTypeInfo;
import com.bff.gaia.streaming.api.xjoin.core.enums.EJoinType;
import com.bff.gaia.streaming.api.xjoin.core.side.cache.AbsSideCache;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Lists;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Maps;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * Description: gaia-parent
 * Created by tiger on 2019/7/18 16:56.
 */
public abstract class SideInfo implements Serializable {

	protected RowTypeInfo rowTypeInfo;

	protected List<FieldInfo> outFieldInfoList;

	protected List<String> equalFieldList = Lists.newArrayList();

	protected List<Integer> equalValIndex = Lists.newArrayList();

	protected String sqlCondition = "";

	protected String sideSelectFields = "";

	protected EJoinType joinType;

	//key:Returns the value of the position, value: the ref field index​in the input table
	protected Map<Integer, Integer> inFieldIndex = Maps.newHashMap();

	//key:Returns the value of the position, value:  the ref field index​in the side table
	protected Map<Integer, Integer> sideFieldIndex = Maps.newHashMap();

	//key:Returns the value of the position, value:  the ref key field index​in the side table
	protected List<Integer> outFieldIndex = new ArrayList<>();
	protected int[] outFieldIndexs;

	//key:Returns the value of the position, value:  the ref field name​in the side table
	protected Map<Integer, String> sideFieldNameIndex = Maps.newHashMap();

	protected SideTableInfo sideTableInfo;

	protected AbsSideCache sideCache;

	protected int projectNum;

	public SideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList,
	                SideTableInfo sideTableInfo) {
		this.rowTypeInfo = rowTypeInfo;
		this.outFieldInfoList = outFieldInfoList;
		this.joinType = joinInfo.getJoinType();
		this.sideTableInfo = sideTableInfo;
		projectNum = outFieldInfoList.size();
		parseSelectFields(joinInfo);
		buildEqualInfo(joinInfo, sideTableInfo);
		buildOutFieldIndex();
	}

	public void parseSelectFields(JoinInfo joinInfo) {
		String sideTableName = joinInfo.getSideTableName();
		String nonSideTableName = joinInfo.getNonSideTable();
		List<String> fields = Lists.newArrayList();

//		int sideIndex = 0;
//		int outFieldIndex=0;
//		for(FieldInfo fieldInfo:outFieldInfoList){
//			if(fieldInfo.getTable().equalsIgnoreCase(sideTableName)){
//				if(fieldInfo.isJoinKey()){
//					dealOneEqualCon(fieldInfo.getFieldName());
//				}else {
//					fields.add(fieldInfo.getFieldName());
//					sideFieldIndex.put(outFieldIndex,sideIndex);
//					sideFieldNameIndex.put(outFieldIndex,fieldInfo.getFieldName());
//					outFieldIndex++;
//				}
//				sideIndex++;
//			}else if (fieldInfo.getTable().equalsIgnoreCase(nonSideTableName)){
//				int nonSideIndex=rowTypeInfo.getFieldIndex(fieldInfo.getFieldName());
//				inFieldIndex.put(outFieldIndex,nonSideIndex);
//				outFieldIndex++;
//			}else {
//				throw new RuntimeException("unknown table " + fieldInfo.getTable());
//			}
//		}

//		int sideIndex = 0;
		for (int i = 0; i < outFieldInfoList.size(); i++) {
			FieldInfo fieldInfo = outFieldInfoList.get(i);
			if (fieldInfo.getTable().equalsIgnoreCase(sideTableName)) {
				if (fieldInfo.isJoinKey()) {
					dealOneEqualCon(fieldInfo.getFieldName(),true);
					projectNum--;
				} else {
					outFieldIndex.add(i);

				}
				fields.add(fieldInfo.getFieldName());
				sideFieldIndex.put(i, sideTableInfo.getFieldList().indexOf(fieldInfo.getFieldName()));
				sideFieldNameIndex.put(i, fieldInfo.getFieldName());
//				sideIndex++;
			} else if (fieldInfo.getTable().equalsIgnoreCase(nonSideTableName)) {
				if(fieldInfo.isJoinKey()){
					dealOneEqualCon(fieldInfo.getFieldName(),false);
				}
				int nonSideIndex = rowTypeInfo.getFieldIndex(fieldInfo.getFieldName());
				outFieldIndex.add(i);
				inFieldIndex.put(i, nonSideIndex);


//				System.out.println(rowTypeInfo);
//				System.out.println(fieldInfo.getFieldName()+" "+inFieldIndex);
			} else {
				throw new RuntimeException("unknown table " + fieldInfo.getTable());
			}
		}


		if (fields.size() == 0) {
			throw new RuntimeException("select non field from table " + sideTableName);
		}

		sideSelectFields = String.join(",", fields);
	}

	public void dealOneEqualCon(String equalFieldName, boolean isSideTable) {

		int equalFieldIndex = -1;
		if (isSideTable) {

			List<String> sideTableFields = sideTableInfo.getFieldList();
			for (int i = 0; i < sideTableFields.size(); i++) {
				if (sideTableFields.get(i).equals(equalFieldName)) {
					equalFieldIndex = i;
					equalFieldList.add(equalFieldName);
					break;
				}
			}
			if (equalFieldIndex == -1) {
				throw new RuntimeException("can't find equal field " + equalFieldName);
			}
		} else {
			equalFieldIndex = rowTypeInfo.getFieldIndex(equalFieldName);
			if (equalFieldIndex == -1) {
				throw new RuntimeException("can't find equal field " + equalFieldName);
			}
			equalValIndex.add(equalFieldIndex);
		}


//		if(leftTableName.equalsIgnoreCase(sideTableName)){
//			equalFieldList.add(leftField);
//			int equalFieldIndex = -1;
//			for(int i=0; i<rowTypeInfo.getFieldNames().length; i++){
//				String fieldName = rowTypeInfo.getFieldNames()[i];
//				if(fieldName.equalsIgnoreCase(rightField)){
//					equalFieldIndex = i;
//				}
//			}
//			if(equalFieldIndex == -1){
//				throw new RuntimeException("can't find equal field " + rightField);
//			}
//
//			equalValIndex.add(equalFieldIndex);
//
//		}else if(rightTableName.equalsIgnoreCase(sideTableName)){
//
//			equalFieldList.add(rightField);
//			int equalFieldIndex = -1;
//			for(int i=0; i<rowTypeInfo.getFieldNames().length; i++){
//				String fieldName = rowTypeInfo.getFieldNames()[i];
//				if(fieldName.equalsIgnoreCase(leftField)){
//					equalFieldIndex = i;
//				}
//			}
//			if(equalFieldIndex == -1){
//				throw new RuntimeException("can't find equal field " + rightField);
//			}
//
//			equalValIndex.add(equalFieldIndex);
//
//		}else{
//			throw new RuntimeException("resolve equalFieldList error:" + sqlNode.toString());
//		}
	}

	public abstract void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo);

	public void buildOutFieldIndex() {
		outFieldIndexs = new int[outFieldIndex.size()];
		for (int i = 0; i < outFieldIndexs.length; i++) {
			outFieldIndexs[i] = outFieldIndex.get(i);
		}
	}

	public RowTypeInfo getRowTypeInfo() {
		return rowTypeInfo;
	}

	public void setRowTypeInfo(RowTypeInfo rowTypeInfo) {
		this.rowTypeInfo = rowTypeInfo;
	}

	public List<FieldInfo> getOutFieldInfoList() {
		return outFieldInfoList;
	}

	public void setOutFieldInfoList(List<FieldInfo> outFieldInfoList) {
		this.outFieldInfoList = outFieldInfoList;
	}

	public List<String> getEqualFieldList() {
		return equalFieldList;
	}

	public void setEqualFieldList(List<String> equalFieldList) {
		this.equalFieldList = equalFieldList;
	}

	public List<Integer> getEqualValIndex() {
		return equalValIndex;
	}

	public void setEqualValIndex(List<Integer> equalValIndex) {
		this.equalValIndex = equalValIndex;
	}

	public String getSqlCondition() {
		return sqlCondition;
	}

	public void setSqlCondition(String sqlCondition) {
		this.sqlCondition = sqlCondition;
	}

	public String getSideSelectFields() {
		return sideSelectFields;
	}

	public void setSideSelectFields(String sideSelectFields) {
		this.sideSelectFields = sideSelectFields;
	}

	public EJoinType getJoinType() {
		return joinType;
	}

	public void setJoinType(EJoinType joinType) {
		this.joinType = joinType;
	}

	public Map<Integer, Integer> getInFieldIndex() {
		return inFieldIndex;
	}

	public void setInFieldIndex(Map<Integer, Integer> inFieldIndex) {
		this.inFieldIndex = inFieldIndex;
	}

	public Map<Integer, Integer> getSideFieldIndex() {
		return sideFieldIndex;
	}

	public void setSideFieldIndex(Map<Integer, Integer> sideFieldIndex) {
		this.sideFieldIndex = sideFieldIndex;
	}

	public SideTableInfo getSideTableInfo() {
		return sideTableInfo;
	}

	public void setSideTableInfo(SideTableInfo sideTableInfo) {
		this.sideTableInfo = sideTableInfo;
	}

	public AbsSideCache getSideCache() {
		return sideCache;
	}

	public void setSideCache(AbsSideCache sideCache) {
		this.sideCache = sideCache;
	}

	public Map<Integer, String> getSideFieldNameIndex() {
		return sideFieldNameIndex;
	}

	public void setSideFieldNameIndex(Map<Integer, String> sideFieldNameIndex) {
		this.sideFieldNameIndex = sideFieldNameIndex;
	}

	public int getProjectNum() {
		return projectNum;
	}

	public void setProjectNum(int projectNum) {
		this.projectNum = projectNum;
	}

	public List<Integer> getOutFieldIndex() {
		return outFieldIndex;
	}

	public void setOutFieldIndex(List<Integer> outFieldIndex) {
		this.outFieldIndex = outFieldIndex;
	}

	public int[] getOutFieldIndexs() {
		return outFieldIndexs;
	}

	public void setOutFieldIndexs(int[] outFieldIndexs) {
		this.outFieldIndexs = outFieldIndexs;
	}
}